/* Orbas:
 *     A open source CORBA Specification implementation from Huihoo.
 *
 * Copyright 2002-2003 Huihoo.org, Inc. All Right Reserved.
 *
 * This software is licensed under LGPL license.
 * See terms of license at gnu.org.
 *
 * For more information, visit:
 *
 * http://www.huihoo.org/orbas
 */

package org.huihoo.orbas.rtutil.threadpool;

/**
 * <p>Description:
 *     Composite the RTPooledExecutor to support the thread pool with lanes or without lanes.
 * </p>
 * <p>[<a href="http://gee.cs.oswego.edu/dl/classes/EDU/oswego/cs/dl/util/concurrent/intro.html">
 *  Introduction to util.concurrent package.
 * </a>]
 * <p>Copyright (c) 2003</p>
 * <p>Company: <a href="http://www.huihoo.org/orbas">huihoo.org</a></p>
 * @author <a href="http://www.huihoo.org/~cocia">Cocia Lin(cocia@163.com)</a>
 * @see <a href="http://www.huihoo.org/orbas">http://www.huihoo.org/orbas</a>
 * @version 1.0
 */

import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;

import EDU.oswego.cs.dl.util.concurrent.Channel;
import EDU.oswego.cs.dl.util.concurrent.Executor;

public class RTThreadPool implements Executor {

	//
	// private PriorityChannel requestQueue;
	// key:ThreadLane;value:Executor
	private Map pool;
	private Object _poollock = new Object();
	// Destoryed flag.
	private boolean destoryed = false;
	// Allow borrow flag.
	private boolean allowBorrow = false;

	/**
	 * e.m. now here 2 5 8 10 priority in the pool. but user want to get the
	 * priority 6 executor. so we should give user the 5,the smallest difference
	 * to 6.
	 * 
	 * @return if the lane pool no active thread to obtain,then will borrow from
	 *         the low lanes. if there is not any active in the all low
	 *         lanes,then will return the near priority lane and will cause the
	 *         request to wait.
	 * 
	 */
	private ThreadLane getNearLaneInPool(int priority) {
		System.out.println("pool::Enter the getNearLaneInPool(...)");
		System.out.println("Param:priority=" + priority);
		synchronized (_poollock) {

			Set keys = pool.keySet();
			// Make SortedSet.
			TreeSet sorted = new TreeSet(new LaneAscendingComparator());
			sorted.addAll(keys);
			Iterator itr = sorted.iterator();
			int smallestDiff = Integer.MAX_VALUE;
			// The near lane.
			ThreadLane retLane = null;
			while (itr.hasNext()) {
				ThreadLane _lane = (ThreadLane) itr.next();
				int _diff = _lane.getPriority() - priority;
				if (Math.abs(_diff) < Math.abs(smallestDiff)) {
					smallestDiff = _diff;
					retLane = _lane;
				}
			}
			ThreadLane _curLane = retLane;
			// Handle for allow borrow.
			if (allowBorrow) {
				// if current thread lane's pool no active thread,then borrow
				// from the low lane.
				RTPooledExecutor _exe = (RTPooledExecutor) pool.get(retLane);
				while (_exe.getPoolSize() <= 0) {

					// Borrow from low lane's pool.
					// Get the low lane.
					Iterator _itr = sorted.iterator();
					Object _pre = null;
					while (_itr.hasNext()) {
						ThreadLane _cur = (ThreadLane) _itr.next();
						if (_cur.getPriority() == retLane.getPriority()) {
							break;
						}
						//
						_pre = _cur;
					}
					//
					if (_pre == null) {
						// current lane's priority is the lowest.
						// No active thread in all lanes.
						// return the current priority lane thread pool and
						// request wait.
						retLane = _curLane;
						break;
					}
					//
					retLane = (ThreadLane) _pre;
					_exe = (RTPooledExecutor) pool.get(retLane);
					// continue to loop.

				}
			}
			// return result.
			System.out.println("Return near priority=" + retLane.getPriority());
			System.out.println("pool::Exit the getNearLaneInPool(...)");
			return retLane;
		}
	}

	private int getCurrentThreadPriority() {
		return 1;

	}

	private void startNewTask(Runnable logic) throws InterruptedException {
		System.out.println("pool::Enter the startNewTask()");
		int priority = this.getCurrentThreadPriority();

		ThreadLane lane = this.getNearLaneInPool(priority);
		Executor executor = (Executor) pool.get(lane);
		// Execute the task.

		// Runnable logic = (Runnable) requestQueue.take();
		executor.execute(logic);

		System.out.println("pool::Exit the startNewTask()");

	}

	public void execute(Runnable command) throws InterruptedException {
		// check state.
		if (destoryed) {
			throw new IllegalStateException(
					"RT Thread pool has been destoryed.");
		}
		//
		System.out.println("pool::Enter the execute(...)");
		synchronized (_poollock) {
			// this.requestQueue.put(command);
			startNewTask(command);
		}
		System.out.println("pool::Exit the execute(...)");

	}

	/**
	 * Initialize all setting.
	 */
	private void init(ThreadLane[] lanes, int requestBufferMax) {
		synchronized (_poollock) {
			this.pool = new HashMap();
			//
			int len = lanes.length;
			for (int i = 0; i < len; i++) {
				Executor executor = this.createPooledExecutor(
						lanes[i].getStaitc_threads(),
						lanes[i].getDynamic_threads(), lanes[i].getPriority(),
						requestBufferMax);
				this.pool.put(lanes[i], executor);
			}
			//
			// this.requestQueue = new PriorityChannel(requestBufferMax);
		}
	}

	//
	public RTThreadPool(int priority, int poolSize) {
		ThreadLane lane = new ThreadLane(priority, poolSize, 0);
		init(new ThreadLane[] { lane }, 1);
	}

	public RTThreadPool(int priority, int poolSize, boolean requestBuffer,
			int requestBufferMax) {
		ThreadLane lane = new ThreadLane(priority, poolSize, 0);
		if (!requestBuffer) {
			requestBufferMax = 1;
		}
		init(new ThreadLane[] { lane }, requestBufferMax);
	}

	public RTThreadPool(ThreadLane[] lanes, boolean allowBorrow,
			boolean requestBuffer, int requestBufferMax) {
		if (!requestBuffer) {
			requestBufferMax = 1;
		}
		// set allow borrow flag.
		this.allowBorrow = allowBorrow;
		// initialize.
		init(lanes, requestBufferMax);
	}

	/**
	 * Destory this thread pool. If there are some task in the queue,will wait
	 * for those ending.
	 */
	public void destory() {
		synchronized (_poollock) {
			Collection _executors = pool.values();
			Iterator itr = _executors.iterator();
			int smallestDiff = Integer.MAX_VALUE;
			// loop to shutdown all the executors.
			while (itr.hasNext()) {
				RTPooledExecutor _executor = (RTPooledExecutor) itr.next();
				// Shutdown After Processing Currently Queued Tasks
				_executor.shutdownAfterProcessingCurrentlyQueuedTasks();
			}
		}
		// Set destoryed flag.
		this.destoryed = true;

	}

	protected Executor createPooledExecutor(int minPoolSize, int maxPoolSize,
			int priority, int requestBufferSize) {
		Channel channel = createChannel(requestBufferSize);
		return new RTPooledExecutor(channel, minPoolSize, maxPoolSize, priority);
	}

	protected Channel createChannel(int maxSize) {
		return new PriorityChannel(maxSize);
	}
	/*
	 * class ExecutorLogic extends RealtimeThread { private Runnable logic;
	 * private boolean busy = false; private Object _runlock = new Object();
	 * private void setBusy(boolean busy) { this.busy = busy; } public boolean
	 * getBusy() { return busy; } public void setRunLogicAndStart(Runnable
	 * logic) { synchronized (_runlock) { this.logic = logic; _runlock.notify();
	 * } } public void setNewPriority(int newPriority) { PriorityParameters
	 * param = (PriorityParameters) this.getSchedulingParameters(); //change
	 * priority. param.setPriority(newPriority); } public int
	 * getCurrentPriority() { PriorityParameters param = (PriorityParameters)
	 * this.getSchedulingParameters(); //get the return param.getPriority(); }
	 * public void run() { synchronized (_runlock) { while (true) { if (logic !=
	 * null) { this.setBusy(true); logic.run(); this.setBusy(false); } //wait
	 * here for new run logic arrive. try { _runlock.wait(); } catch
	 * (InterruptedException e) { e.printStackTrace(); } } } } }
	 */

}